package com.foreveross.mofang.datapersistence.application.impl;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.inject.Inject;
import javax.inject.Named;

import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.foreveross.mofang.datapersistence.application.HbaseDataOperationApplication;
import com.foreveross.mofang.datapersistence.application.vo.FileInfo;
import com.foreveross.mofang.datapersistence.application.vo.HbaseQueryVo;
import com.foreveross.mofang.datapersistence.infra.IHbaseRepository;
import com.foreveross.mofang.datapersistence.infra.service.impl.HbaseMapToJsonMap;
import com.foreveross.mofang.datapersistence.infra.service.impl.JsonMapToHbaseMap;

/**
 * hbase操作类基础应用层
 * 
 * @author fb
 */
@Named
public class HbaseDataOperationApplicationImpl implements HbaseDataOperationApplication {

	private static final Logger logger = LoggerFactory.getLogger(HbaseDataOperationApplicationImpl.class);

	@Inject
	private IHbaseRepository hbaseRepository;
	
	public void createTable(String tableName, List<String> familys) {
		try {
			hbaseRepository.createTable(tableName, familys);
		} catch (Exception e) {
			logger.error("hbase创建表{}, familys：{}时出错", new Object[] { tableName, familys, e});
		}

	}

	@SuppressWarnings({"unchecked" })
	public void addRecord(String tableName, Map<String, Object> data) throws Exception {
		Map<String, Object> hbaseMap = new HashMap<String, Object>();
		String rowKey = data.get("rowKey").toString();
		
		new JsonMapToHbaseMap((Map<String, Object>) data.get("data"), hbaseMap).translation();
		hbaseRepository.addRecord(tableName, rowKey, hbaseMap);
	}
	
	@SuppressWarnings("unchecked")
	public void batchAddRecord(String tableName, List<Map<String, Object>> datas) throws Exception {
		Map<String, Map<String, Object>> rowDatas = new HashMap<String, Map<String, Object>>();
		Map<String, Object> hbaseMap = null;
		
		for (Map<String, Object> data : datas) {
			hbaseMap = new HashMap<String, Object>();
			new JsonMapToHbaseMap((Map<String, Object>) data.get("data"), hbaseMap).translation();
			rowDatas.put(data.get("rowKey").toString(), hbaseMap);
		}
		
		hbaseRepository.batchAddRecord(tableName, rowDatas);
	}

	public void delOneRecord(String tableName, String rowKey) throws Exception {
		hbaseRepository.delOneRecord(tableName, rowKey);
	}

	public void delMultRecord(String tableName, String... rowKeys)
			throws Exception {
		hbaseRepository.delMultRecord(tableName, rowKeys);
	}

	public Map<String, Object> getRecord(String tableName, String rowKey)
			throws Exception {
		return new HbaseMapToJsonMap(new HashMap<String, Object>(), hbaseRepository.getRecord(tableName, rowKey)).translation();
	}

	public List<Map<String, Object>> queryRecords(String tableName, HbaseQueryVo queryParams, int pageSize)
			throws Exception {
		List<Filter> rowFilters = new ArrayList<Filter>();
		List<Map<String, Object>> result = new ArrayList<Map<String,Object>>();
		
		if (queryParams != null && queryParams.getHbaseQuery() != null) {
			for (HbaseQueryVo.HbaseQuery query : queryParams.getHbaseQuery()) {
				rowFilters.add(new SingleColumnValueFilter(Bytes.toBytes(query.getFamily().trim()), 
						Bytes.toBytes(query.getQualifier()),
						CompareOp.valueOf(query.getCompareOp().trim()),
						Bytes.toBytes(query.getValue().trim())));
			}
		}
		
		rowFilters.add( new PageFilter(pageSize));
		
		for (Map<String, Object> hbaseMap : hbaseRepository.queryRecords(tableName, new FilterList(rowFilters))) {
			result.add(new HbaseMapToJsonMap(new HashMap<String, Object>(), hbaseMap).translation());
		}
		
		return result;
	}

	public void saveDataFileInfo(FileInfo fileInfo) {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		Map<String, Object> hbaseMap = new HashMap<String, Object>();
		String family = "FILEINFO|";
		hbaseMap.put(family + "name", fileInfo.getOriginalName());
		hbaseMap.put(family + "download", fileInfo.getDownloadUrl());
		hbaseMap.put(family + "md5", fileInfo.getMd5());
		hbaseMap.put(family + "size", fileInfo.getFileSize());
		hbaseMap.put(family + "fileCreateDate", fileInfo.getCreateDate());
		hbaseMap.put(family + "createDate", sdf.format(new Date()));
		
		try {
			hbaseRepository.addRecord("DATA_FILE_INFO", fileInfo.getOriginalName(), hbaseMap);
		} catch (IOException e) {
			logger.error("下载的数据文件信息入库出错:{}", fileInfo, e);
		}
	}
	
}
